package com.sunday.common.mq.rabbit.study.base.e13_PriorityQueueSupport;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * https://www.rabbitmq.com/priority.html
 * <p>
 * 优先队列支持
 * <p>
 * RabbitMQ 从版本 3.5.0 开始在核心中实现了优先级队列。
 * 可以使用客户端提供的可选参数将任何队列转换为优先级队列（但与使用可选参数而不是策略的其他功能不同）。
 * 该实现支持有限数量的优先级：255。建议使用 1 到 10 之间的值。
 * <p>
 * 优先队列支持
 * 若要声明优先级队列，请使用 x-max-priority 可选队列参数。 此参数应为 1 到 255 之间的正整数， 指示队列应支持的最大优先级
 * 然后，发布商可以使用 basic.properties 的优先级字段发布优先消息。数字越大表示越高 优先权
 * <p>
 * 为什么不支持策略定义
 * 为队列定义可选参数的最便捷方法是通过策略。 策略是配置 TTL、队列长度限制和 其他可选队列参数。
 * 但是，策略不能用于配置优先级，因为策略是动态的 并且可以在声明队列后进行更改。优先级队列永远无法更改优先级的数量 队列声明后的支持，因此策略不是安全的使用选项。
 * <p>
 * 消息优先级字段定义为 无符号字节，因此在实践中优先级应介于 0 之间 和255。
 * 没有优先级属性的消息被视为 如果它们的优先级为 0。优先级为 高于队列的最大值被视为 以最高优先级发布。
 * <p>
 * 优先级和资源使用情况的最大数量
 * 如果需要优先级队列，我们建议使用 1 到 10 之间的队列。 当前使用更多优先级将通过使用更多的 Erlang 进程来消耗更多的 CPU 资源。运行时调度也会受到影响。
 * <p>
 * 与其他功能的交互
 * 一般来说，优先级队列具有标准的所有功能 RabbitMQ 队列：它们支持持久性、分页、镜像、 等等。开发人员应该进行一些交互 意识到的。
 * 应过期的邮件仍将 仅从队列的头部过期。这意味着与 对于普通队列，即使是每个队列的 TTL 也可能导致过期 优先级较低的邮件卡在未过期的邮件后面 优先级更高的。这些消息将永远不会传递， 但它们将显示在队列统计信息中。
 * 具有最大长度集的队列将像往常一样从 队列以强制实施限制。这意味着更高的优先级 消息可能会被删除，以便为优先级较低的消息让路， 这可能不是您所期望的。
 * <p>
 * 消息。服务器和应用程序之间传送的数据，本质上就是一段数据，由Properties和Payload(body)组成。
 * <p>
 * https://www.rabbitmq.com/publishers.html#message-properties
 * Property	        类型                 描述	                                                        必填？
 * Delivery mode	枚举（1 或 2）	    2 表示“持久”，1 表示“瞬态”。某些客户端库公开此属性 作为布尔值或枚举。	    是的
 * Type	            字符串	            特定于应用程序的消息类型，例如“订单.已创建”	                        不
 * Headers	        映射（字符串 => 任意）	具有字符串标头名称的任意标头映射	                                    不
 * Content type	    字符串	            内容类型，例如“应用程序/json”。由应用程序使用，而不是核心 RabbitMQ	    不
 * Content encoding	字符串	            内容编码，例如“gzip”。由应用程序使用，而不是核心 RabbitMQ	            不
 * Message ID	    字符串	            任意消息 ID	                                                    不
 * Correlation ID	字符串	            帮助将请求与响应相关联，请参阅教程 6	                                不
 * Reply To	        字符串	            携带响应队列名称，请参阅教程 6	                                    不
 * Expiration	    字符串	            每条消息的 TTL	                                                不
 * Timestamp	    时间戳	            应用程序提供的时间戳	                                            不
 * User ID	        字符串	            用户 ID，如果已设置，则已验证	                                    不
 * App ID	        字符串	            应用程序名称	                                                    不
 * <p>
 * 如何保证消息的不丢失，三个地方做到持久化。
 * <p>
 * Exchange需要持久化。
 * Queue需要持久化。
 * Message需要持久化。
 */
@Slf4j
public class PriorityQueueSupport {


    /**
     * https://www.rabbitmq.com/priority.html
     * <p>
     * 队列优先级与消息优先级的比较
     * <p>
     * 优先级和资源使用情况的最大数量
     * 如果需要优先级队列，我们建议使用 1 到 10 之间的队列。 当前使用更多优先级将通过使用更多的 Erlang 进程来消耗更多的 CPU 资源。运行时调度也会受到影响。
     */
    @Test
    @SneakyThrows
    public void queueAndMessagePriority() {

        String queue = UUID.randomUUID().toString();
        log.info("本次测试队列 : {}", queue);

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();
            /**
             * 优先级和资源使用情况的最大数量 x-max-priority
             * 如果需要优先级队列，我们建议使用 1 到 10 之间的队列。 当前使用更多优先级将通过使用更多的 Erlang 进程来消耗更多的 CPU 资源。运行时调度也会受到影响。
             */
            channel.queueDeclare(queue, false, false, false, Collections.singletonMap("x-max-priority", 5));

            for (int i = 0; i <= 10; i++) {
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .priority(i)        //优先级
                        .build();
                channel.basicPublish("", queue, props, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            channel.basicConsume(queue, true,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" consumerTag : {} ", consumerTag);
                        log.info(" envelope : {} ", delivery.getEnvelope());
                        log.info(" properties : {} ", delivery.getProperties());
                        log.info(" [x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            /**
             * 本轮测试可知 :
             * 1. 队列不设置优先级，参数设置，将会已FIFO的规则进行
             * 2. 队列设置优先级，a. 参数超过优先级，将会已超过部分先进行消费 FIFO规则 b. 参数未超过优先级，将已优先级规则进行消费
             */
            TimeUnit.SECONDS.sleep(2);

        }

    }


    /**
     * 已知
     * 没有 priority 属性的消息被视为 如果它们的优先级为 0。优先级为 高于队列的最大值被视为 以最高优先级发布。
     * 那么如果没有 priority 的队列直接结果是什么，答案也是不起作用的
     */
    @Test
    @SneakyThrows
    public void messagePriorityNotQueuePriority() {

        String queue = UUID.randomUUID().toString();

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel channel = connection.createChannel();

            channel.queueDeclare(queue, false, false, false, null);

            for (int i = 0; i <= 10; i++) {
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .priority(i)        //优先级
                        .build();
                channel.basicPublish("", queue, props, String.valueOf(i).getBytes(StandardCharsets.UTF_8));
            }

            channel.basicConsume(queue, true,
                    (consumerTag, delivery) -> {
                        String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                        log.info(" consumerTag : {} ", consumerTag);
                        log.info(" envelope : {} ", delivery.getEnvelope());
                        log.info(" properties : {} ", delivery.getProperties());
                        log.info(" [x] Received '{}'", message);
                    },
                    consumerTag -> {
                    });

            TimeUnit.SECONDS.sleep(2);

        }

    }

}
